package dataconsume;

import org.apache.kafka.clients.consumer.*;
import java.sql.*;
import java.time.Duration;
import java.util.Collections;
import java.util.Properties;
import java.util.regex.Pattern;

public class MathQuestionConsumer {

    // MySQL配置
    private static final String JDBC_URL = "jdbc:mysql://localhost:3306/db_teaching?useUnicode=true&characterEncoding=UTF-8";
    private static final String JDBC_USER = "root";
    private static final String JDBC_PASSWORD = "20040429lk";

    // Kafka配置
    private static final String KAFKA_TOPIC = "javaTopic";
    private static final String GROUP_ID = "math-question-consumer-group";

    public static void main(String[] args) {
        // 初始化Kafka消费者
        Properties kafkaProps = new Properties();
        kafkaProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        kafkaProps.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
        kafkaProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        kafkaProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        kafkaProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // 初始化数据库连接
        try (Connection conn = DriverManager.getConnection(JDBC_URL, JDBC_USER, JDBC_PASSWORD);
             Consumer<String, String> consumer = new KafkaConsumer<>(kafkaProps)) {

            // 创建存储表
            createTableIfNotExists(conn);

            // 订阅主题
            consumer.subscribe(Collections.singletonList(KAFKA_TOPIC));

            // 消费消息
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));
                for (ConsumerRecord<String, String> record : records) {
                    processRecord(conn, record.value());
                }
                consumer.commitSync();
            }
        } catch (SQLException | RuntimeException e) {
            System.err.println("系统异常: " + e.getMessage());
            e.printStackTrace();
        }
    }

    private static void createTableIfNotExists(Connection conn) throws SQLException {
        String sql = "CREATE TABLE IF NOT EXISTS tb_math_questions (" +
                "id INT PRIMARY KEY, " +
                "question TEXT NOT NULL, " +
                "answer VARCHAR(255) NOT NULL, " +
                "difficulty_level VARCHAR(50) NOT NULL, " +
                "topic VARCHAR(100) NOT NULL) " +
                "CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci";

        try (Statement stmt = conn.createStatement()) {
            stmt.execute(sql);
            System.out.println("数据表已就绪");
        }
    }

    private static void processRecord(Connection conn, String csvLine) {
        // 使用正则表达式解析带引号的CSV
        Pattern csvPattern = Pattern.compile(",(?=([^\"]*\"[^\"]*\")*[^\"]*$)");
        String[] fields = csvPattern.split(csvLine, 5);

        try {
            // 去除字段的引号
            for (int i = 0; i < fields.length; i++) {
                fields[i] = fields[i].replaceAll("^\"|\"$", "");
            }

            // 准备插入语句
            String insertSql = "INSERT INTO tb_math_questions VALUES (?, ?, ?, ?, ?) " +
                    "ON DUPLICATE KEY UPDATE question=VALUES(question), answer=VALUES(answer)";

            try (PreparedStatement pstmt = conn.prepareStatement(insertSql)) {
                pstmt.setInt(1, Integer.parseInt(fields[0]));
                pstmt.setString(2, fields[1]);
                pstmt.setString(3, fields[2]);
                pstmt.setString(4, fields[3]);
                pstmt.setString(5, fields[4]);

                pstmt.executeUpdate();
                System.out.println("插入记录: ID=" + fields[0]);
            }
        } catch (SQLException e) {
            System.err.println("数据库操作失败: " + e.getMessage());
        } catch (NumberFormatException | ArrayIndexOutOfBoundsException e) {
            System.err.println("数据格式错误: " + csvLine);
        }
    }
}